package com.diao.flink.streamapi.transform

import com.diao.flink.streamapi.source.SensorReading
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
/**
 * @author: Chenzhidiao
 * @date: 2020/3/27 11:20
 * @description:split和select算子结合使用
 * @version: 1.0
 */
object SplitAndSelect {
//  def main(args: Array[String]): Unit = {
//    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//    env.setParallelism(1)
//    val stream: DataStream[SensorReading] = env.readTextFile("E:\\Workspace\\flink\\src\\main\\resources\\sensorFile").map(data=>{
//      //将每条数据按逗号切分，然后封装成SensorReading对象
//      val dataArray = data.split(",")
//      SensorReading(dataArray(0).trim,dataArray(1).trim.toLong,dataArray(2).trim.toDouble)
//    })
//    //split和select算子结合使用
//    val splitStream: SplitStream[SensorReading] = stream.split(t=>{
//      if (t.temperature>69)Seq("high")else Seq("low")
//    })
//    val high: DataStream[SensorReading] = splitStream.select("high")
//    val low: DataStream[SensorReading] = splitStream.select("low")
//    val all: DataStream[SensorReading] = splitStream.select("high","low")
//
//    high.print()
//
//    env.execute()
//  }

}
